package bsp.diff;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.sync.SyncException;

import util.DiffComputation;
import util.DiffMessage;
import util.DiffMessage2;
import util.Parsable;
import util.ScanDList;
import util.Utilties;

public class DiffStepTwo<T0 extends Parsable<T0>, T1 extends Parsable<T1>, T2 extends Parsable<T2>>
		extends DiffMapper<T0, T1, T2, DiffMessage<T1>> {

	T2 accSum;
	T0 insT0;

	T2 pacc;
//	Utilties<T1> tl1;
//	Utilties<T2> tl2;

	public DiffStepTwo(DiffComputation<T0, T1, T2> di) {
		super(di);

		otimsAcc = otimes.id();
		accSum = oplus.id();
		pacc = null;

		insT0 = p.getInstanceOfT0();
 

	}

	@Override
	public void mapAll(BSPPeer<LongWritable, Text, Text, NullWritable, Text> contex)
			throws IOException, SyncException, InterruptedException {

		LongWritable key = new LongWritable(-1);
		Text line = new Text();
		ScanDList<T1> ptimes = new ScanDList<T1>();
		DiffMessage<T1> receiver;

		// get all messages
		int countMsg = 0;

		Text msg = contex.getCurrentMessage();

		System.out.println("+++++DiffStepTwo DEBUG received++++++ " + msg);

		receiver = new DiffMessage<T1>(msg.toString(), otimsAcc.getClass());
		while (msg != null && receiver != null) {

			countMsg++;
			T1 previousAcc = otimes.id();
			// msg =received.getMsg();
			System.out.println(" msg  =" + msg);
			try {
				previousAcc = receiver.parse(msg.toString()).getSecond(); // @@@
			} catch (NoSuchMethodException | SecurityException
					| ClassNotFoundException | InstantiationException
					| IllegalAccessException | IllegalArgumentException
					| InvocationTargetException e) {
				e.printStackTrace();
			}
			if (previousAcc == null) {
				throw new InterruptedException("parse failed");

			} else {
				// System.out.println("DiffMessage: " + received.getMsgId()
				if (previousAcc != null)
					ptimes.put(receiver.getMsgId(), previousAcc);
			}

			System.out.println("countMsg =" + countMsg);
			msg = contex.getCurrentMessage();

		}

		// do scan
		if (countMsg > 0 && ptimes != null) {
			T1 preAllAcc = Utilties.AccumPrevousTimes(chunkey, ptimes,
					otimes);

			System.out.println("preAllAcc: " + preAllAcc);
			this.otimsAcc = otimes.evaluate(preAllAcc, this.otimsAcc); // otimes.evaluate(cons_c,
			if (chunkey == 0)
				this.otimsAcc = otimes.evaluate(cons_c, otimsAcc);
		}

		T2 result = oplus.id();

		while (contex.readNext(key, line)) {

			StringTokenizer itr = new StringTokenizer(line.toString().trim(),
					TOKEN);
			T0 thisOne;
			while (itr.hasMoreTokens()) {
				// sum += Integer.parseInt(itr.nextToken());
				String tmp = itr.nextToken();
				thisOne = insT0.parse(tmp);
				pacc = p.evaluate(thisOne, otimsAcc);
				accSum = oplus.evaluate(accSum, pacc);
				otimsAcc = otimes.evaluate(otimsAcc, q.evaluate(thisOne));

			}
		}
		// ///////////////////////////////////////////////////////////////////////////
		// if (ptimes.lastKey() == chunkey){
		// result = oplus.evaluate(accSum, g.evaluate(otimsAcc));
		//
		//
		// contex.write(new Text(result.toString()), NullWritable.get());
		// } else {
		DiffMessage2<T1, T2> sendmsg = new DiffMessage2<T1, T2>(chunkey,
				otimsAcc, accSum);
		addMesg(contex.getPeerName(numPeers - 1), new Text(sendmsg.toString()));
		System.out.println("send msg: " + sendmsg);
		// }

	}

	// in place update list1
	// private void zipList(ScanDList<T2> list1,
	// final ScanDList<Pair<T0, T0>> list2,
	// DiffComputation<T0, T1, T2> diff) throws InterruptedException {
	// long key = list1.firstKey(); // from first chunk-key, i.e., 0
	// long higher = -1;
	// while (list1.higherKey(key) != null) {
	// higher = list1.higherKey(key);
	//
	// Pair<T0, T0> eFsteLst = list2.get(key);
	// Pair<T0, T0> nextEFsteLst = list2.get(higher);
	//
	// T1 qx0 = diff.q.evaluate(eFsteLst.getSecond());
	// T1 cqx0 = diff.otimes.evaluate(diff.cons_c, qx0);
	// T2 px1cqx0 = diff.p.evaluate(nextEFsteLst.getFirst(), cqx0);
	// T2 fixed = diff.oplus.evaluate(px1cqx0, list1.get(key));
	// list1.put(higher, fixed);
	// key = higher;
	// }
	// }
}
